#include "FlexThreadPool.h"
#include <mutex>
#include <memory>
#include <functional>
#include <assert.h>
#include <algorithm>
#include "BlockingQueue.h"

#define DEBUG_OUTPUT(format, ...) printf(format "\n", __VA_ARGS__)

namespace Ark
{
    static const int TASK_DEFAULT = 1;
    static const int TASK_EXIT = 2;

    //----- FlexThreadPool -----
    FlexThreadPool::FlexThreadPool()
        :_busyThreadCount(0)
    {
    }

    FlexThreadPool::~FlexThreadPool()
    {
        Stop();
    }

    void FlexThreadPool::Start(unsigned minThreadCount, unsigned maxThreadCount, unsigned idleSecondsToRecycle)
    {
        std::lock_guard<std::mutex> lock(_mutex);
        if (_stop || !_threads.empty()) {
            return;
        }

        _minThreadCount = minThreadCount == 0 ? std::thread::hardware_concurrency() : minThreadCount;
        _maxThreadCount = maxThreadCount == 0 ? std::thread::hardware_concurrency() : maxThreadCount;
        _maxThreadCount = (std::max)(_minThreadCount, _maxThreadCount);
        _idleSecondsToRecycle = idleSecondsToRecycle == 0 ? DEFAULT_RECYCLE_IDLE_TIME : idleSecondsToRecycle;
        
        for (unsigned i = 0; i < _minThreadCount; ++i) {
            _threads.emplace_back(new std::thread(std::bind(&self::_WorkLoop, this, _idleSecondsToRecycle)));
        }
        DEBUG_OUTPUT("threadpool starts, min=%u, max=%u, idle=%u", _minThreadCount, _maxThreadCount, _idleSecondsToRecycle);
    }

    void FlexThreadPool::Stop()
    {
        {
            std::lock_guard<std::mutex> lock(_mutex);
            if (_stop) {
                return;
            }
            _stop = true; // prevent adding any more task
        }
        
        // post exit task to let thread exit
        _PostExitTask();    

        // wait all threads exit
        auto tickBeginStopThread = std::chrono::high_resolution_clock::now();
        DEBUG_OUTPUT("threadpool begins stopping");
        for (auto thread : _threads) {
            thread->join();
        }
        auto usedTime = std::chrono::high_resolution_clock::now() - tickBeginStopThread;
        DEBUG_OUTPUT("threadpool finishes stopping, used-time=%u ms", (unsigned)std::chrono::duration_cast<std::chrono::milliseconds>(usedTime).count());
    }

    void FlexThreadPool::_PostExitTask()
    {
        std::shared_ptr<BlockingQueue::Task> task(new BlockingQueue::Task);
        task->userdata = (void*)TASK_EXIT;
        task->task = [] {
            // do nothing
        };
        for (unsigned i = 0; i < _maxThreadCount; ++i) {
            _taskQueue.Push(task);
        }
    }

    bool FlexThreadPool::Post(const std::function<void()>& callback)
    {
        BlockingQueue::TaskPtr task(new BlockingQueue::Task(callback));
        task->userdata = (void*)TASK_DEFAULT;
        return _Post(task);
    }

    bool FlexThreadPool::_Post(BlockingQueue::TaskPtr task)
    {
        std::lock_guard<std::mutex> lock(_mutex);
        if (_stop) {
            return false;
        }

        if (_busyThreadCount >= _threads.size()
            && _threads.size() < _maxThreadCount) {
            _threads.emplace_back(new std::thread(std::bind(&self::_WorkLoop, this, _idleSecondsToRecycle)));
            DEBUG_OUTPUT("add new thread to pool, count=%u/%u", (unsigned)(_threads.size()), _maxThreadCount);
        }
        _taskQueue.Push(task);
        return true;
    }

    void FlexThreadPool::_WorkLoop(int idleSeconds)
    {
        DEBUG_OUTPUT("work thread starts, idle=%d", idleSeconds);
        while (true)
        {
            auto task = _taskQueue.Pop(idleSeconds);
            if (!task) { //timeout, check this thread should recycle
                if (_RecycleThread()) {
                    return;
                }
                continue;
            }

            if (task->userdata == (void*)TASK_EXIT) {
                DEBUG_OUTPUT("thread exits due to threadpool is exiting");
                return;
            }

            ++_busyThreadCount;
            task->task();
            --_busyThreadCount;
        }
    }

    bool FlexThreadPool::_RecycleThread()
    {
        std::lock_guard<std::mutex> lock(_mutex);
        if (_stop) { 
            return false; // in stopping progress, flex thread may be in join state, we should not detach it
        }

        if (_threads.size() <= _minThreadCount) {
            return false;
        }

        for (auto it = _threads.begin(); it != _threads.end(); ++it) {
            auto& thread = *it;
            if (thread->get_id() == std::this_thread::get_id()) {
                thread->detach();
                _threads.erase(it);
                DEBUG_OUTPUT("recycle thread, thread-count=%u", (unsigned)_threads.size());
                return true;
            }
        }
        assert(!"recycle thread not found");
        return true;
    }

}